import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.concurrent.ArrayBlockingQueue;

public class Sequencer  {
	
	private User user = null;
	private long sequenceID = 0; 
	private Hashtable<String, NormalMessage> received = new Hashtable<>();
	
	private ArrayBlockingQueue<Message> queue = new ArrayBlockingQueue<>(1024);
	
	public User getUser() {
		return user;
	}

	public void setUser(User user) {
		this.user = user;
	}

	public ArrayBlockingQueue<Message> getQueue() {
		return queue;
	}

	public void addQueue(Message msg) {
		this.queue.add(msg);
	}

	public Sequencer(User user) {
		this.user = user;
	}
	
	public static void main(String[] args) throws Exception {
		String name = args[0];
		User user = Configuration.getInstance().getUserByName(name);
		Sequencer sequencer = new Sequencer(user);
		new SequencerListener(user, sequencer);
	}
	
	public void bDeliver(NormalMessage msg) throws InterruptedException {
		if (!received.containsKey(msg.getSender()+"@"+msg.getSenderID())) {
			System.out.println("=-------"+msg.getSender()+"@"+msg.getSenderID());
			for(User r : Configuration.getInstance().getUsers()) {
				if (!r.getName().equalsIgnoreCase(this.user.getName())) {
					TotalMessage totalMsg = new TotalMessage(msg.getContent(), this.getUser(), msg.getSender(), sequenceID);
					new Sender(totalMsg, r, queue);
				}
			}
			sequenceID++;
			received.put(msg.getSender()+"@"+msg.getSenderID(),msg);
		}
	}
	
}

class SequencerListener implements Runnable {
	
	private User user = null;
	private Sequencer sequencer = null;
	
	public SequencerListener(User user, Sequencer sequencer) {
		this.user = user;
		this.sequencer = sequencer;
		new Thread(this, user.getName()+"").start();
	}
	
	@SuppressWarnings("resource")
	@Override
	public void run() {
		DatagramSocket serverSocket = null;
		try {
			serverSocket = new DatagramSocket(user.getPort());
		} catch (SocketException e) {
			e.printStackTrace();
		}
		byte[] receiveData = new byte[1024];
		while (true) {
			DatagramPacket receivePacket = new DatagramPacket(receiveData,
					receiveData.length);
			try {
				serverSocket.receive(receivePacket);
			} catch (IOException e) {
				e.printStackTrace();
			}
			ObjectInputStream in;
			Message msg = null;
			try {
				in = new ObjectInputStream(new ByteArrayInputStream(receiveData)); 
				msg = (Message) in.readObject();
				in.close();
			} catch (IOException e) {
				e.printStackTrace();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			} 
			
			if (msg instanceof AckMessage) {
				sequencer.addQueue((AckMessage)msg);
			} else {
				try {
					sequencer.bDeliver((NormalMessage)msg);
					System.out.println("*****Sequencer delivers " + ((NormalMessage)msg).getSender() + " " + ((NormalMessage)msg).getSenderID());
					new Sender(new AckMessage(msg.getSenderID()), ((NormalMessage)msg).getSender(), sequencer.getQueue());
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		
	}
}
